package se.sics.kompics.tutorial.tutorial1.appPFD;

import java.util.HashSet;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import se.kth.ict.id2203.flp2p.FairLossPointToPointLink;
import se.kth.ict.id2203.flp2p.Flp2pSend;
import se.kth.ict.id2203.pp2p.PerfectPointToPointLink;
import se.kth.ict.id2203.pp2p.Pp2pSend;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Kompics;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.address.Address;
import se.sics.kompics.timer.ScheduleTimeout;
import se.sics.kompics.timer.Timer;
import se.sics.kompics.tutorial.tutorial1.appPFD.crash.CrashEvent;
import se.sics.kompics.tutorial.tutorial1.appPFD.messages.Flp2pMessage;
import se.sics.kompics.tutorial.tutorial1.appPFD.messages.HeartBeatMessage;
import se.sics.kompics.tutorial.tutorial1.appPFD.messages.Pp2pMessage;

public class AppPFD extends ComponentDefinition {
	
	Positive<PerfectPointToPointLink> pp2p = requires(PerfectPointToPointLink.class);
	Positive<FairLossPointToPointLink> flp2p = requires(FairLossPointToPointLink.class);
	Positive<Timer> timer = requires(Timer.class);
	Negative<PerfectFailureDetector> info = provides(PerfectFailureDetector.class);
	/**
	 * For PFD algo 1
	 */
	private static final long heartBeatDelay = 1000;
	private static final long transmissionDelay = 4000;

	private static final Logger logger = LoggerFactory
			.getLogger(AppPFD.class);

	//private String[] commands;
	//private int lastCommand;
	private Set<Address> neighborSet;
	private Address self;
	/**
	 * For PFD algo 1
	 */
	private Set<Address> aliveSet;
	private Set<Address> detectedSet;

	/**
	 * Instantiates a new application1.
	 */
	public AppPFD() {
		subscribe(handleInit, control);
		//subscribe(handleStart, control);
		//subscribe(handleContinue, timer);
		subscribe(handlePp2pMessage, pp2p);
		subscribe(handleFlp2pMessage, flp2p);
		/**
		 * For PFD algo 1
		 */
		subscribe(handleReceivedHeartBeatMessage, pp2p);
		subscribe(handleApplicationHeartBeat, timer);
		subscribe(handleCheck, timer);
	}

	Handler<AppPFD1Init> handleInit = new Handler<AppPFD1Init>() {
		public void handle(AppPFD1Init event) {
			//commands = event.getCommandScript().split(":");
			//lastCommand = -1;
			neighborSet = event.getNeighborSet();
			self = event.getSelf();
			
			/**
			 * For PFD algo 1
			 */
			aliveSet = new HashSet<Address>();
			aliveSet.addAll(event.getNeighborSet());
			detectedSet = new HashSet<Address>();
			
			ScheduleTimeout heartBeatTimeout = new ScheduleTimeout(heartBeatDelay);
			heartBeatTimeout.setTimeoutEvent(new AppPFDHeartBeat(heartBeatTimeout));
			trigger(heartBeatTimeout, timer);
			
			ScheduleTimeout checkTimeout = new ScheduleTimeout(heartBeatDelay + transmissionDelay);
			checkTimeout.setTimeoutEvent(new AppPFDCheck(checkTimeout));
			trigger(checkTimeout, timer);
		}
	};

	/*Handler<Start> handleStart = new Handler<Start>() {
		public void handle(Start event) {
			doNextCommand();
		}
	};

	Handler<AppPFDContinue> handleContinue = new Handler<AppPFDContinue>() {
		public void handle(AppPFDContinue event) {
			doNextCommand();
		}
	};*/
	
	/**
	 * For PFD algo 1
	 */
	Handler<AppPFDHeartBeat> handleApplicationHeartBeat = new Handler<AppPFDHeartBeat>() {
		public void handle(AppPFDHeartBeat event) {
			doHeartBeat();
		}
	};
	
	Handler<AppPFDCheck> handleCheck = new Handler<AppPFDCheck>() {
		public void handle(AppPFDCheck event) {
			doCheck();
		}
	};

	Handler<Pp2pMessage> handlePp2pMessage = new Handler<Pp2pMessage>() {
		public void handle(Pp2pMessage event) {
			logger.info("Received perfect message {}", event.getMessage());
		}
	};
	
	/**
	 * For PFD algo 1
	 */
	Handler<HeartBeatMessage> handleReceivedHeartBeatMessage = new Handler<HeartBeatMessage>() {
		public void handle(HeartBeatMessage event) {
			//logger.info("Received HeartBeat message {} from {}", event.getMessage(), event.getSource());
			aliveSet.add(event.getSource());
		}
	};

	Handler<Flp2pMessage> handleFlp2pMessage = new Handler<Flp2pMessage>() {
		public void handle(Flp2pMessage event) {
			logger.info("Received lossy message {}", event.getMessage());
		}
	};

	/*private final void doNextCommand() {
		lastCommand++;

		if (lastCommand > commands.length) {
			return;
		}
		if (lastCommand == commands.length) {
			logger.info("DONE ALL OPERATIONS");
			Thread applicationThread = new Thread("ApplicationThread") {
				public void run() {
					BufferedReader in = new BufferedReader(
							new InputStreamReader(System.in));
					while (true) {
						try {
							String line = in.readLine();
							doCommand(line);
						} catch (Throwable e) {
							e.printStackTrace();
						}
					}
				}
			};
			applicationThread.start();
			return;
		}
		String op = commands[lastCommand];
		doCommand(op);
	}

	private void doCommand(String cmd) {
		if (cmd.startsWith("P")) {
			doPerfect(cmd.substring(1));
			doNextCommand();
		} else if (cmd.startsWith("L")) {
			doLossy(cmd.substring(1));
			doNextCommand();
		} else if (cmd.startsWith("S")) {
			doSleep(Integer.parseInt(cmd.substring(1)));
		} else if (cmd.startsWith("X")) {
			doShutdown();
		} else if (cmd.equals("help")) {
			doHelp();
			doNextCommand();
		} else {
			logger.info("Bad command: '{}'. Try 'help'", cmd);
			doNextCommand();
		}
	}

	private final void doHelp() {
		logger.info("Available commands: P<m>, L<m>, S<n>, help, X");
		logger.info("Pm: sends perfect message 'm' to all neighbors");
		logger.info("Lm: sends lossy message 'm' to all neighbors");
		logger.info("Sn: sleeps 'n' milliseconds before the next command");
		logger.info("help: shows this help message");
		logger.info("X: terminates this process");
	}*/

	@SuppressWarnings("unused")
	private final void doPerfect(String message) {
		for (Address neighbor : neighborSet) {
			logger.info("Sending perfect message {} to {}", message, neighbor);
			trigger(new Pp2pSend(neighbor, new Pp2pMessage(self, message)),
					pp2p);
		}
	}

	@SuppressWarnings("unused")
	private final void doLossy(String message) {
		for (Address neighbor : neighborSet) {
			logger.info("Sending lossy message {} to {}", message, neighbor);
			trigger(new Flp2pSend(neighbor, new Flp2pMessage(self, message)),
					flp2p);
		}
	}
	
	/**
	 * For PFD algo 1
	 */
	private final void doHeartBeat() {
		HeartBeatMessage heartBeatMessage = new HeartBeatMessage(self);
		for (Address neighbor : neighborSet) {
			//logger.info("Sending heartbeat message {} to {}", heartBeatMessage.getMessage(), neighbor);
			trigger(new Pp2pSend(neighbor, heartBeatMessage), pp2p);
		}
		
		ScheduleTimeout heartBeatTimeout = new ScheduleTimeout(heartBeatDelay);
		heartBeatTimeout.setTimeoutEvent(new AppPFDHeartBeat(heartBeatTimeout));
		trigger(heartBeatTimeout, timer);
	}
	
	private final void doCheck() {
		//logger.info("Checking if there is any crashed neighbor...");
		for (Address neigbor : neighborSet) {
			if (!aliveSet.contains(neigbor) && !detectedSet.contains(neigbor)) {
				detectedSet.add(neigbor);
				//crash event
				trigger(new CrashEvent(neigbor), info);
				logger.info("Neighbor {} crashed!", neigbor);
			}
		}
		aliveSet = new HashSet<Address>();
		ScheduleTimeout checkTimeout = new ScheduleTimeout(heartBeatDelay + transmissionDelay);
		checkTimeout.setTimeoutEvent(new AppPFDCheck(checkTimeout));
		trigger(checkTimeout, timer);
	}

	@SuppressWarnings("unused")
	private void doSleep(long delay) {
		logger.info("Sleeping {} milliseconds...", delay);

		ScheduleTimeout st = new ScheduleTimeout(delay);
		st.setTimeoutEvent(new AppPFDContinue(st));
		trigger(st, timer);
	}

	@SuppressWarnings("unused")
	private void doShutdown() {
		System.out.println("2DIE");
		System.out.close();
		System.err.close();
		Kompics.shutdown();
	}

}
